package com.harboursoftware.xstorage.fs;

import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.Ostermiller.util.MD5OutputStream;
import com.google.common.collect.ListMultimap;
import com.harboursoftware.xstorage.XStorageException;
import com.harboursoftware.xstorage.ac.AccessControlPolicy;
import com.harboursoftware.xstorage.db.BucketDao;
import com.harboursoftware.xstorage.model.XBucket;
import com.harboursoftware.xstorage.model.XObject;
import com.harboursoftware.xstorage.model.XUser;

/**
 * {@link XFileSystem} 的 <a href="http://hadoop.apache.org/hdfs/">HDFS</a>实现，在HDFS中，每个{@link XBucket}
 * 对应一个HDFS中的目录/${bucket.name} 而{@link XObject}则对应于Bucket目录下的一个文件/${bucket.name}/{object.key}.
 * <p>
 * {@link XBucket}的具体信息保存在数据库中，通过{@link BucketDao}访问。
 * </p>
 * 
 * @since 0.1
 * @author Simon Leung
 * 
 */
public class HDFSFileSystem implements XFileSystem {
    private final static Logger log = LoggerFactory.getLogger(HDFSFileSystem.class);
    private FileSystem hadoopFileSystem;
    private BucketDao bucketDao;

    public HDFSFileSystem(FileSystem hadoopFileSystem, BucketDao bucketDao) {
        this.hadoopFileSystem = hadoopFileSystem;
        this.bucketDao = bucketDao;
    }

    @Override
    public void createBucket(XBucket bucket) throws BucketAlreadyOwnedByOtherUserException, XStorageException {
        XUser creator = bucket.getOwner();
        try {
            AccessControlPolicy persistentedAcp = getBucketAcp(bucket);
            if (persistentedAcp.isOwner(creator)) { // already own by creator
                log.debug("bucket [{}] already own by same user[{}]", bucket.getName(), creator.getId());
                // updates acp
                if (!persistentedAcp.equals(bucket.getAcp())) {
                    putBucketAcp(bucket);
                }
            } else { // already own by other
                throw new BucketAlreadyOwnedByOtherUserException("bucket [" + bucket.getName() + "] already exist.");
            }
        } catch (BucketNotExistException e) {
            // bucket not exist, create new one
            createBucketAtHDFS(bucket);
        }
    }

    private void createBucketAtHDFS(XBucket bucket) throws XStorageException {
        Path path = getPath(bucket);
        // TODO:处理事务
        try {
            hadoopFileSystem.mkdirs(path);
            bucketDao.saveBucket(bucket);
        } catch (IOException e) {
            throw new XStorageException(e);
        }
    }

    @Override
    public void listObjects(XBucket bucket, String delimiter, String marker, int maxKeys, String prefix)
        throws BucketNotExistException, XStorageException {
        if (!isBucketExisted(bucket)) {
            throw new BucketNotExistException("bucket[" + bucket.getName() + "] does not exists");
        }
        Path bucketPath = getPath(bucket);
        FileStatus[] fileStatus = null;
        try {
            fileStatus = hadoopFileSystem.listStatus(bucketPath);
        } catch (IOException e) {
            throw new XStorageException(e);
        }

        bucket.setSize(fileStatus.length);

        List<XObject> objcets = new ArrayList<XObject>();
        maxKeys = maxKeys > fileStatus.length ? fileStatus.length : maxKeys;
        for (int i = 0; i < maxKeys; i++) {
            FileStatus status = fileStatus[0];
            XObject object = fileStatus2Object(status, bucket);
            
            this.loadObjectACP(object);
            this.loadObjectEtag(object);
            objcets.add(object);
        }
        
        bucket.setObjects(objcets);
        // TODO : 未完成
    }
    
    private XObject fileStatus2Object(FileStatus status, XBucket bucket) {
        XObject object = new XObject();
        object.setBucket(bucket);
        String pathString = status.getPath().toString();
        log.debug("path : {}" , pathString);
        
        String bucketPathString = getPathString(bucket);
        int startIndex = pathString.indexOf(bucketPathString, 1) + bucketPathString.length() + 1;
        String key = pathString.substring(startIndex);
        log.debug("key : {}", key);
        object.setKey(key);
        object.setLastModified(status.getModificationTime());
        object.setLength(status.getLen());
        return object;
    }

    @Override
    public void putBucketAcp(XBucket bucket) throws BucketNotExistException, XStorageException {
        if (!isBucketExisted(bucket)) {
            throw new BucketNotExistException("bucket [" + bucket.getName() + "] does not exist");
        }
        XBucket persistentedBucket = bucketDao.findBucketByName(bucket.getName());
        persistentedBucket.setAcp(bucket.getAcp());
        bucketDao.updateBucket(persistentedBucket);
    }

    @Override
    public void deleteBucket(XBucket bucket) throws BucketNotExistException, BucketNotEmptyException, XStorageException {
        if (!isBucketExisted(bucket)) {
            throw new BucketNotExistException("bucket [" + bucket.getName() + "] does not exist");
        }

        if (!isBucketEmpty(bucket)) {
            throw new BucketNotEmptyException("bucket [" + bucket.getName() + "] is not empty");
        }

        // TODO : 处理事务
        try {
            boolean successful = hadoopFileSystem.delete(getPath(bucket), false);
            if (!successful) {
                throw new XStorageException("fail to delete bucket[" + bucket.getName() + "]");
            }
        } catch (IOException e) {
            throw new XStorageException(e);
        }
        bucketDao.deleteBucket(bucket);
    }

    @Override
    public boolean isBucketEmpty(XBucket bucket) throws BucketNotExistException, XStorageException {
        Path path = getPath(bucket);
        FileStatus[] status = null;
        try {
            status = hadoopFileSystem.listStatus(path);
        } catch (IOException e) {
            throw new XStorageException(e);
        }
        int totalFiles = status.length;
        return totalFiles == 0;
    }

    @Override
    public boolean isBucketExisted(XBucket bucket) throws XStorageException {
        Path path = getPath(bucket);
        try {
            return hadoopFileSystem.exists(path);
        } catch (IOException e) {
            throw new XStorageException(e);
        }
    }

    // public boolean isBucketExisted(String bucketName) throws IOException {
    // Path path = new Path(Path.SEPARATOR + bucketName);
    // return hadoopFileSystem.exists(path);
    // }
    @Override
    public void loadBucketACP(XBucket bucket) throws BucketNotExistException, XStorageException {
        bucket.setAcp(getBucketAcp(bucket));
    }

    public AccessControlPolicy getBucketAcp(XBucket bucket) throws BucketNotExistException, XStorageException {
        XBucket persistentedBucket = bucketDao.findBucketByName(bucket.getName());
        if (persistentedBucket == null) {
            throw new BucketNotExistException("bucket [" + bucket.getName() + "] does not exist");
        }
        return persistentedBucket.getAcp();
    }

    @Override
    public String createObject(XObject object) throws XStorageException {
        // data format : acp(AccessControlPolicy) + headers(Map<String, String[]>) + data(InputStream))
        // save object
        Path path = getPath(object);
        InputStream input = object.getInputStream();
        // save etag
        // TODO: 考试是否可以使用getChecksum()来实现 etag
        String md5 = null;
        try {
            OutputStream rawOutput = hadoopFileSystem.create(path, true);
            ObjectOutputStream objectOutput = new ObjectOutputStream(rawOutput);
            objectOutput.writeObject(object.getAcp());
            objectOutput.writeObject(object.getHeaders());
            MD5OutputStream md5Output = new MD5OutputStream(objectOutput);
            IOUtils.copy(input, md5Output);
            rawOutput.flush();
            IOUtils.closeQuietly(input);
            md5Output.close();
            objectOutput.close();
            rawOutput.close();

            md5 = md5Output.getHashString();
            Path etagFilePath = getEtagFilePath(object);
            OutputStream md5FileOutput = hadoopFileSystem.create(etagFilePath, true);
            md5FileOutput.write(md5.getBytes());
            md5FileOutput.flush();
            md5FileOutput.close();
        } catch (IOException e) {
            throw new XStorageException(e);
        }

        return md5;
    }

    /**
     * 不包含etag TODO: 应该是采用：getObject(XObject object) 或是 getObject(String bucket, String key) 或是loadObject()
     * 
     * @see #createObject
     */
    @SuppressWarnings("unchecked")
    @Override
    public void loadObject(XObject object) throws BucketNotExistException, ObjectNotExistException, XStorageException {
        // data format : acp(AccessControlPolicy) + headers(Map<String, String[]>) + data(InputStream))
        if (!this.isBucketExisted(object.getBucket())) {
            throw new BucketNotExistException("bucket[" + object.getBucket().getName() + "] does not exists");
        }

        if (!this.isObjectExisted(object)) {
            throw new ObjectNotExistException("object[" + object.getKey() + "] on bucket["
                + object.getBucket().getName() + "] does not exists");
        }
        // TODO: object.setKey() && object.setBucket()

        Path path = getPath(object);
        ObjectInputStream objectInput = null;
        AccessControlPolicy acp = null;
        ListMultimap<String, String> headers = null;
        FileStatus status = null;
        try {
            InputStream rawInput = hadoopFileSystem.open(path);
            objectInput = new ObjectInputStream(rawInput);
            acp = (AccessControlPolicy) objectInput.readObject();
            headers = (ListMultimap<String, String>) objectInput.readObject();
            status = hadoopFileSystem.getFileStatus(path);
        } catch (IOException e) {
            throw new XStorageException(e);
        } catch (ClassNotFoundException e) {
            throw new XStorageException("unknown error occured", e);
        }
        object.setLastModified(status.getModificationTime());
        object.setLength(status.getLen());
        object.setAcp(acp);
        object.setHeaders(headers);
        object.setInputStream(objectInput);
    }

    @Override
    public void deleteObject(XObject object) throws ObjectNotExistException, XStorageException {
        Path path = getPath(object);
        Path etagFilePath = getEtagFilePath(object);
        try {
            // TODO : 处理事务
            boolean successful1 = hadoopFileSystem.delete(etagFilePath, false);
            boolean successful2 = hadoopFileSystem.delete(path, false);
            boolean allSuccessful = successful1 && successful2;
            if (!allSuccessful) {
                throw new XStorageException("fail to delete object[" + object + "]");
            }
        } catch (IOException e) {
            throw new XStorageException(e);
        }
    }

    @Override
    public boolean isObjectExisted(XObject object) throws XStorageException {
        Path path = getPath(object);
        try {
            return hadoopFileSystem.exists(path);
        } catch (IOException e) {
            throw new XStorageException(e);
        }
    }

    public boolean isObjectExisted(String bucket, String key) throws IOException {
        Path path = new Path(Path.SEPARATOR + bucket + key);
        return hadoopFileSystem.exists(path);
    }

    @Override
    public void loadObjectACP(XObject object) throws ObjectNotExistException, XStorageException {
        Path path = getPath(object);
        AccessControlPolicy acp = null;
        try {
            InputStream rawInput = hadoopFileSystem.open(path);
            ObjectInputStream objectInput = new ObjectInputStream(rawInput);
            acp = (AccessControlPolicy) objectInput.readObject();
            objectInput.close();
            rawInput.close();
        } catch (IOException e) {
            throw new XStorageException(e);
        } catch (ClassNotFoundException e) {
            throw new XStorageException("unknown error occured", e);
        }

        object.setAcp(acp);
    }

    @Override
    public void loadObjectEtag(XObject object) throws BucketNotExistException, ObjectNotExistException,
        XStorageException {
        // load etag
        Path etagFilePath = getEtagFilePath(object);
        try {
            InputStream etagFileInput = hadoopFileSystem.open(etagFilePath);
            String etag = IOUtils.toString(etagFileInput);
            object.setEtag(etag);
            etagFileInput.close();
        } catch (IOException e) {
            throw new XStorageException(e);
        }
    }

    public void setHadoopFileSystem(FileSystem hadoopFileSystem) {
        this.hadoopFileSystem = hadoopFileSystem;
    }

    public FileSystem getHadoopFileSystem() {
        return hadoopFileSystem;
    }

    public BucketDao getBucketDao() {
        return bucketDao;
    }

    public void setBucketDao(BucketDao bucketDao) {
        this.bucketDao = bucketDao;
    }

    private Path getPath(XObject object) {
        String key = object.getKey();
        String pathString = null;
        if (key.startsWith("/")) {
            pathString = getPathString(object.getBucket()) + object.getKey();
        } else {
            pathString = getPathString(object.getBucket()) + "/" + object.getKey();
        }
        return new Path(pathString);
    }

    private Path getEtagFilePath(XObject object) {
        String key = object.getKey();
        String pathString = null;
        if (key.startsWith("/")) {
            pathString = getPathString(object.getBucket()) + object.getKey() + ".etag.x";
        } else {
            pathString = getPathString(object.getBucket()) + "/" + object.getKey() + ".etag.x";
        }
        return new Path(pathString);
    }

    private String getPathString(XBucket bucket) {
        return Path.SEPARATOR + bucket.getName();
    }

    private Path getPath(XBucket bucket) {
        return new Path(getPathString(bucket));
    }

}
